package com.example.sql;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

/**
 * Author wangJinLong
 * Date 2025/8/22 16:09
 **/
public class FlinkSqlJoin {
    public static void main(String[] args) {
        lookupJoin();

    }

    private static void lookupJoin() {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment environment = TableEnvironment.create(settings);

        generateData(environment);

        environment.executeSql("CREATE TEMPORARY TABLE customers (\n" +
                "  id STRING,\n" +
                "  name STRING,\n" +
                "  country STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:mysql://106.54.174.109:13306/my_cdc',\n" +
                "  'table-name' = 'customers',\n" +
                "  'driver' = 'com.mysql.cj.jdbc.Driver',\n" +
                "  'username' = 'root',\n" +
                "  'password' = 'mysql@123456'\n" +
                ");");

        environment.executeSql("SELECT o.order_id, o.customer_id, c.name, c.country, o.amount, o.order_time\n" +
                "FROM orders AS o\n" +
                "INNER JOIN customers AS c\n" +
                "ON o.customer_id = c.id;").print();
    }

    private static void temporalJoin() {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment environment = TableEnvironment.create(settings);

        generateData(environment);

        environment.executeSql("SELECT orders.*, products.price\n" +
                "FROM orders\n" +
                "LEFT JOIN products FOR SYSTEM_TIME AS OF orders.order_time\n" +
                "ON orders.product_id = products.id;").print();
    }

    private static void intervalJoin() {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment environment = TableEnvironment.create(settings);

        generateData(environment);

        environment.executeSql("SELECT o.*, s.ship_id\n" +
                "FROM orders o, shipments s\n" +
                "WHERE o.order_id = s.order_id\n" +
                "AND o.order_time BETWEEN s.ship_time - INTERVAL '10' SECOND AND s.ship_time;").print();
    }

    private static void leftJoin() {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment environment = TableEnvironment.create(settings);

        generateData(environment);

        environment.executeSql("SELECT *, (orders.amount * products.price) as sum_price\n" +
                "FROM orders\n" +
                "LEFT JOIN products\n" +
                "ON orders.product_id = products.id;").print();
    }

    private static void innerJoin() {
        EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        TableEnvironment environment = TableEnvironment.create(settings);

        generateData(environment);

        environment.executeSql("SELECT *, (orders.amount * products.price) as sum_price\n" +
                "FROM orders\n" +
                "INNER JOIN products\n" +
                "ON orders.product_id = products.id;").print();
    }

    private static void generateData(TableEnvironment environment) {
        environment.executeSql("CREATE TABLE products (\n" +
                    "id STRING,\n" +
                    "price DECIMAL(5, 2),\n" +
                    "bidtime TIMESTAMP_LTZ(3),\n" +
                    "WATERMARK FOR bidtime AS bidtime,\n" +
                    "PRIMARY KEY(id) NOT ENFORCED\n" +
                ")\n" +
                "WITH (\n" +
                    "'connector'='datagen',\n" +
                    "'rows-per-second' = '1',\n" +
                    "'number-of-rows' = '20',\n" +
                    "'fields.id.length' = '1'\n" +
                ");");

        environment.executeSql("CREATE TABLE orders (\n" +
                "order_id STRING,\n" +
                "product_id STRING,\n" +
                "customer_id STRING,\n" +
                "amount INT,\n" +
                "order_time TIMESTAMP_LTZ(3),\n" +
                "WATERMARK FOR order_time AS order_time - INTERVAL '10' SECOND\n" +
                ")\n" +
                "WITH (\n" +
                "'connector'='datagen',\n" +
                "'rows-per-second' = '1',\n" +
                "'fields.order_id.length' = '2',\n" +
                "'fields.product_id.length' = '1',\n" +
                "'fields.customer_id.length' = '1',\n" +
                "'fields.amount.min' = '1',\n" +
                "'fields.amount.max' = '10'\n" +
                ");");

        environment.executeSql("CREATE TABLE shipments (\n" +
                "ship_id STRING,\n" +
                "order_id STRING,\n" +
                "ship_time  TIMESTAMP_LTZ(3),\n" +
                "WATERMARK FOR ship_time AS ship_time - INTERVAL '10' SECOND\n" +
                ")\n" +
                "WITH (\n" +
                "'connector'='datagen',\n" +
                "'rows-per-second' = '1',\n" +
                "'fields.ship_id.length' = '10',\n" +
                "'fields.order_id.length' = '2'\n" +
                ");");
    }
}
